package com.sunlands.deskmate.message.handler.sender.schedule;

import com.sunlands.deskmate.channel.ChannelInfo;
import com.sunlands.deskmate.channel.ChannelManager;
import com.sunlands.deskmate.message.enums.ProtocolEnum;
import com.sunlands.deskmate.message.handler.sender.trigger.TriggerEvent;
import com.sunlands.deskmate.message.handler.sender.trigger.TriggerEventEnum;
import com.sunlands.deskmate.message.handler.sender.trigger.SendMessageTrigger;
import com.sunlands.deskmate.message.handler.storager.unsync.UnsyncMessageStorager;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.core.ValueOperations;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Component;
import org.springframework.util.CollectionUtils;
import java.util.List;
import java.util.Objects;
import java.util.Set;
import java.util.concurrent.TimeUnit;

/**
 * 多节点的情况下, 选择其中一个节点做一些单线程的工作
 *
 * @author anjunling@sunlands.com
 * @author liude
 */
@SuppressWarnings("SpringJavaAutowiredFieldsWarningInspection")
@Slf4j
@Component
public class Discoverer {

    private static final String LEADER_KEY = "discover_leader";

    /**
     * 必须设置, 每个节点设置不同的值
     */
    @Value("${WORKER_ID}")
    String workerId;

    @Autowired
    private UnsyncMessageStorager storager;

    @Autowired
    private RedisTemplate<String, String> redisTemplate;

    @Autowired
    private SendMessageTrigger sendDispatcher;

    @Scheduled(cron = "0 0/1 * * * ?")
    public void discoverChannelAll() {

        //选主, 多节点的情况下, 只有一个节点执行
        if (!fastElection()) {
            return;
        }

        //所有未同步唯一标识
        List<String> unsyncKeys = storager.unsyncKeys();

        if (CollectionUtils.isEmpty(unsyncKeys)) {
            return;
        }

        log.debug("unsyncKeys={}", unsyncKeys);

        for (String key : unsyncKeys) {
            String sessionKey = key.replace(UnsyncMessageStorager.MESSAGE_UNSYNC, "");

            ChannelInfo channelInfo = ChannelInfo.parseSessionKey(sessionKey);

//            ClientInfoManager.ClientInfo clientInfo = clientInfoManager.clientInfo(channelInfo.getClientId(), channelInfo.getUserIdentity());

            log.debug("channelInfo={}, clientInfo={}", channelInfo);

//            if (clientInfo == null) {
//                continue;
//            }

            //筛选短连接通道
            dispatch(channelInfo, key, ProtocolEnum.HTTP.getType(), ProtocolEnum.HTTPS.getType());
        }
    }


    @Scheduled(cron = "0/10 * * * * ?")
    public void discoverChannelWebsocket() {
        //当前节点存在长连接通道
        Set<String> sessionKeys = ChannelManager.getAllSessionKeys();

        if (CollectionUtils.isEmpty(sessionKeys)) {
            return;
        }

        log.debug("sessionKeys={}", sessionKeys);

        for (String sessionKey : sessionKeys) {
            ChannelInfo channelInfo = ChannelInfo.parseSessionKey(sessionKey);
//            ClientInfoManager.ClientInfo clientInfo = clientInfoManager.clientInfo(channelInfo.getClientId(), channelInfo.getUserIdentity());
//            if (clientInfo == null) {
//                continue;
//            }

            dispatch(channelInfo, sessionKey, ProtocolEnum.WEBSOCKET.getType(), ProtocolEnum.WEBSOCKETS.getType());
        }
    }

    /**
     * 抢占式选取, 30s过期
     *
     * @return boolean
     */
    private boolean fastElection() {
        ValueOperations<String, String> valueOperations = redisTemplate.opsForValue();
        Boolean result = valueOperations.setIfAbsent(LEADER_KEY, workerId);

        if (Objects.isNull(result)) {
            return false;
        }

        if (result) {
            result = redisTemplate.expire(LEADER_KEY, 30, TimeUnit.SECONDS);
        }

        return !Objects.isNull(result) && result;
    }


    private void dispatch(ChannelInfo channelInfo, String key, String... protocols) {

        if (protocols == null || protocols.length <= 0) {
            return;
        }

//        boolean flag = false;
//        for (String protocol : protocols) {
//            if (protocol.equalsIgnoreCase(clientInfo.getCallbackProtocol())) {
//                flag = true;
//            }
//        }
//        if (!flag) {
//            log.debug("discover protocol not exists {} {}", key, protocols);
//            return;
//        }

        //未同步消息数
        String sessionKey = channelInfo.getSessionKey();
        long size = storager.countUnreadMessage(sessionKey);


        if (size > 0) {

            log.debug("sessionKey={}, countUnreadMessage={}", sessionKey, size);

            //FIXME msgId ?
            TriggerEvent dispatchEvent = new TriggerEvent(TriggerEventEnum.UNSYNC, channelInfo);
            boolean result = sendDispatcher.offer(dispatchEvent);

            log.debug("offer unsync event, event={}, key={}, result={}", dispatchEvent, key, result);
        }

    }

}
